package async.callback;

import cn.hutool.core.lang.Console;
import cn.hutool.core.thread.ThreadUtil;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.*;
import lombok.Builder;
import lombok.Data;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.junit.Test;

import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @author jiangwz
 * @create 2021/6/11.
 */
public class GuavaListeningFutureTest {

    final static ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
    private static final CountDownLatch LATCH = new CountDownLatch(1);
    public static long markTime = System.currentTimeMillis();

    @Data
    @Builder
    static class Order {
        private String orderNo;
    }

    @Test
    public void test01(){
        ListenableFuture<Object> failedFuture = Futures.immediateFailedFuture(new NullPointerException());

        ListenableFuture<List<Order>> listListenableFuture = Futures.immediateFuture(searchOrders());
        Futures.addCallback(listListenableFuture, new FutureCallback<List<Order>>() {

            @Override
            public void onSuccess(List<Order> msg) {
                msg.forEach(System.out::println);
                LATCH.countDown();
            }

            @Override
            public void onFailure(Throwable throwable) {
                LATCH.countDown();
            }
        }, service);

        await();
    }


    @Test
    public void testSimple() {

        startMarkTime();
        ListenableFuture<List<Order>> future = service.submit(() -> searchOrders());
        Futures.addCallback(future, new FutureCallback<List<Order>>() {

            @Override
            public void onSuccess(List<Order> msg) {
                LATCH.countDown();
            }

            @Override
            public void onFailure(Throwable throwable) {

            }
        }, service);

        await();
    }

    @Test
    public void second(){
        startMarkTime();
        ListenableFuture<List<List<Order>>> allAsList = Futures.allAsList(
                service.submit(() -> searchOrders()),
                service.submit(() -> searchOrders()),
                service.submit(() -> searchOrders())
        );

        Futures.addCallback(allAsList, new FutureCallback<List<List<Order>>>() {

            @Override
            public void onSuccess(List<List<Order>> msg) {
                LATCH.countDown();
            }

            @Override
            public void onFailure(Throwable throwable) {

            }
        }, service);

        await();
    }

    public void startMarkTime() {
        markTime = System.currentTimeMillis();
    }

    public void await() {
        try {
            LATCH.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Console.log("cost time={}ms", System.currentTimeMillis() - markTime);
    }

    public List<Order> searchOrders() {
        ThreadUtil.safeSleep(3000);
        return Lists.newArrayList(
                Order.builder().orderNo("1111").build()
        );
    }

    public List<Order> searchOrdersThrowsException() {
        ThreadUtil.safeSleep(3000);
        throw new NullPointerException();
    }

    public List<Long> searchOrders2() {
        ThreadUtil.safeSleep(3000);
        return Lists.newArrayList(
                1L,2L
        );
    }

    @Test
    public void test03(){
        ListenableFuture<List<Order>> listenableFuture = service.submit(() -> searchOrders());
        ListenableFuture<List<Optional<Order>>> transform = Futures.transform(listenableFuture, orders -> {
            List<Optional<Order>> result = Lists.newArrayList();
            orders.forEach(order -> {
                if (order != null) {
                    result.add(Optional.of(order));
                } else {
                    result.add(Optional.empty());
                }
            });
            return result;
        }, MoreExecutors.directExecutor());

        Futures.addCallback(transform, new FutureCallback<List<Optional<Order>>>() {
            @Override
            public void onSuccess(@Nullable List<Optional<Order>> result) {

            }

            @Override
            public void onFailure(Throwable t) {

            }
        }, MoreExecutors.directExecutor());
    }

    @Test
    public void testTimeout(){
        // 默认执行3秒
        ListenableFuture<List<Order>> listenableFuture = service.submit(() -> searchOrders());
        ScheduledExecutorService timeoutExecutor = Executors.newSingleThreadScheduledExecutor();
        ListenableFuture<List<Order>> withTimeout = Futures.withTimeout(listenableFuture, 1, TimeUnit.SECONDS, timeoutExecutor);
        Futures.addCallback(withTimeout, new FutureCallback<List<Order>>() {
            @Override
            public void onSuccess(@Nullable List<Order> result) {
                LATCH.countDown();
                System.out.println(result);
            }

            @Override
            public void onFailure(Throwable t) {
                LATCH.countDown();
                System.out.println("error -> "+t);
            }
        }, MoreExecutors.directExecutor());
        await();

    }

    @Test
    public void testSettableFuture01(){
//        CompletableFuture<List<Order>> completableFuture = CompletableFuture.supplyAsync(() -> searchOrdersThrowsException());
        CompletableFuture<List<Order>> completableFuture = CompletableFuture.supplyAsync(() -> searchOrders());
        SettableFuture<List<Order>> settableFuture = SettableFuture.create();
        completableFuture.whenComplete((orders, exception) -> {
            if (exception != null) {
                settableFuture.setException(exception);
            }else {
                settableFuture.set(orders);
            }
        });

        // 转换
        ListenableFuture<List<String>> transform = Futures.transform(settableFuture, orders -> orders.stream().filter(Objects::nonNull)
                .map(Order::getOrderNo).collect(Collectors.toList()), service);

        FutureCallback<List<String>> callback = new FutureCallback<List<String>>() {
            @Override
            public void onSuccess(@Nullable List<String> result) {
                System.out.println(result);
                LATCH.countDown();
            }

            @Override
            public void onFailure(Throwable t) {
                System.out.println(t);
                LATCH.countDown();
            }
        };

        AsyncCallbackTemplate.withCallback(transform, callback::onSuccess, callback::onFailure, service);

        System.out.println("------------- start ------------------------");
        await();
    }





}
